package intensivecomp.util;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;

import java.io.IOException;

/**
 * 继承RecordReader
 * 该类用来将分片分割成记录，从而生成key和value。例如TextInputFormat中的key和value就是RecordReader的子类产生的。
 * 在这里，我们继承了这个类，将重写了key和value的生成算法。对一个分片来说，只生成一个key-value对。其中key为空，value为该分片
 * 的所有内容
 * @author Xiaoye
 */
public class WholeSplitRecordReader extends RecordReader<Text, Text> {
    // 用来盛放传递过来的分片
    private CustomerFileSplit fileSplit;
    private Configuration conf;
    //将作为key-value中的value值返回
    private Text value = new Text();
    private long pos;
    private Text key = new Text();
    // 因为只生成一条记录，所以只需要调用一次。因此第一次调用过后将processed赋值为true，从而结束key和value的生成
    private boolean processed = false;

    /**
     * 设置RecordReader的分片和配置对象。
     */
    @Override
    public void initialize(InputSplit split, TaskAttemptContext context)
            throws IOException, InterruptedException {
        this.fileSplit = (CustomerFileSplit) split;
        this.conf = context.getConfiguration();
        pos = this.fileSplit.getStart();
    }

    /**
     * 核心算法
     * 用来产生key-value值
     * 将生成的value值存入value对象中
     */
    @Override
    public boolean nextKeyValue() throws IOException, InterruptedException {
        if (!processed) {
            /*
             * 注意这儿，fileSplit中只是存放着待处理内容的位置 大小等信息，并没有实际的内容
             * 因此这里会通过fileSplit找到待处理文件，然后再读入内容到value中
             */
            byte[] contents = new byte[(int) fileSplit.getLength()];
            Path file = fileSplit.getPath();
            FileSystem fs = file.getFileSystem(conf);
            FSDataInputStream in = null;
            try {
                in = fs.open(file);
                if (pos > 0)
                    IOUtils.skipFully(in, pos+1);
                IOUtils.readFully(in, contents, 0, contents.length);
                value.set(contents, 0, contents.length);
            } finally {
                IOUtils.closeStream(in);
            }
            processed = true;
            this.key.set(fileSplit.getId());
            return true;
        }
        return false;
    }

    @Override
    public Text getCurrentKey() throws IOException,
            InterruptedException {
        return key;
    }

    @Override
    public Text getCurrentValue() throws IOException,
            InterruptedException {
        return value;
    }

    @Override
    public float getProgress() throws IOException, InterruptedException {
        return processed ? 1.0f : 0.0f;
    }

    @Override
    public void close() throws IOException {
        //do nothing
    }
}